package com.sink;


import com.Connection.JDBCConnectionTools;
import com.Pojo.IISLog;
import com.Pojo.ServerLog;
import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;


//Flink使用的类都需要实现序列化，进行网络传输
@AllArgsConstructor
@NoArgsConstructor
public class MetricBeatSink implements Serializable {

    private String sqltype;
    private static final Logger logger = LoggerFactory.getLogger(MetricBeatSink.class);



    //解析器必须使用FilebeatJSONKeyValueNginxDeserializationSchema才能使用该方法获取sink
    public SinkFunction<String> GetMetricSinkString(String user, String password, String DBip, String databaseName
    ) throws Exception {

        SinkFunction<String> sink = JdbcSink.sink(
                "insert  into ServerLog (host,datetime,Metric,result,module,Context,version)" +
                        "values" +
                        " (?,?,?,?,?,?,?)",
                new JdbcStatementBuilder<String>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, String s) throws SQLException {
                        ServerLog log = JSONObject.parseObject(s,ServerLog.class);
                        logger.info(log.toString());
                        Timestamp localDateTimeToTimeStamp = Timestamp.valueOf(log.getDatetime());
                        preparedStatement.setString(1, log.getHost());    //preparedStatement只接收date和Timestamp，只能将LocalDateTime转为Timestamp再写入DB
                        preparedStatement.setTimestamp(2, localDateTimeToTimeStamp);
                        preparedStatement.setString(3, log.getMetric());
                        preparedStatement.setString(4, log.getResult());
                        preparedStatement.setString(5, log.getModule());
                        preparedStatement.setString(6, log.getContext());
                        preparedStatement.setString(7, log.getVersion());

                    }
                },
                JdbcExecutionOptions.builder()
                        .withMaxRetries(3)  //插入发生异常重试次数，Flink checkpoint has started
                        .withBatchSize(1000) //批处理最大条数，设置大一些，生产环境的访问频率高
                        .withBatchIntervalMs(500) //每次批量写入最小间隔时间
                        .build(),//只要满足其中一条，JDBC批处理就会执行
//                new JDBCConnectionTools().GetMssqlConnect(user,password,DBip,databaseName)
                new JDBCConnectionTools(this.sqltype).GetConnect(user,password,DBip,databaseName)
        );

        return sink;

    }
}
